package rattlesnake.callback.util;

import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.multipart.*;
import io.netty.util.CharsetUtil;
import org.apache.http.HttpResponse;
import org.apache.http.NameValuePair;
import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.methods.HttpUriRequest;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.nio.reactor.IOReactorException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import rattlesnake.callback.config.CallbackConfig;
import rattlesnake.callback.config.impl.CallbackProperties;
import rattlesnake.callback.http.RequestParser;
import rattlesnake.callback.http.ResponseStatus;
import rattlesnake.callback.modle.dto.HttpRequestWrapper;
import rattlesnake.callback.server.filter.MessageFilter;
import java.io.*;
import java.util.*;
import static io.netty.buffer.Unpooled.copiedBuffer;
import static io.netty.handler.codec.http.HttpHeaders.Names.*;
import static io.netty.handler.codec.http.HttpHeaders.Names.SET_COOKIE;

/**
 * @Package: rattlesnake.callback.util
 * @Description: 异步请求工具
 * @author: liuxin
 * @date: 17/4/12 上午9:09
 */
public class HttpAsyncUtil {
    private static final Logger logger = LoggerFactory.getLogger(HttpAsyncUtil.class);
    private static ConnectingIOReactor ioReactor = null;
    private static PoolingNHttpClientConnectionManager cm = null;
    private static CloseableHttpAsyncClient httpAsyncClient = null;
    private static boolean isFilter = false;

    static {
        try {
            ioReactor = new DefaultConnectingIOReactor();
            cm = new PoolingNHttpClientConnectionManager(ioReactor);
            cm.setMaxTotal(100);
            httpAsyncClient = HttpAsyncClients.custom().setConnectionManager(cm).build();
            isFilter = CallbackProperties.getBoolean(CallbackConfig.ISFILTER);
        } catch (IOReactorException ie) {
            throw new RuntimeException(ie);
        }
        httpAsyncClient.start();
    }

    /**
     * 处理请求头
     *
     * @param request
     * @param baseMeth
     */
    private static void pretreatentHead(HttpRequest request, HttpUriRequest baseMeth) {
        HttpHeaders httpHeaders = request.headers();
        for (Map.Entry<String, String> head : httpHeaders.entries()) {
            if (head.getKey().equals(CONTENT_LENGTH)) {
                continue;
            }
            baseMeth.setHeader(head.getKey(), head.getValue());
            logger.debug("请求头:{}={}", head.getKey(), head.getValue());

        }
    }

    /**
     * 预处理请求头信息
     *
     * @param ctx
     * @param request
     * @param url
     * @param success
     * @param error
     */
    public static void doPostExecute(ChannelHandlerContext ctx, HttpRequest request, HttpObject msg, String url, String success, String error) {
        logger.debug("===THIS IS HTTP POST===");
        boolean cacheFlag = true;
        HttpPost httpPost = new HttpPost(url);
        pretreatentHead(request, httpPost);
        Map<String, String> params = RequestParser.parse(request, msg);
        List<NameValuePair> nvps = new ArrayList<NameValuePair>();
        Set<String> keys = params.keySet();
        for (String key : keys) {
            String value = params.get(key);
            logger.debug("key:{},value:{}", key, value);
            nvps.add(new BasicNameValuePair(key, value));
        }
        try {
            httpPost.setEntity(new UrlEncodedFormEntity(nvps));
        } catch (UnsupportedEncodingException ue) {

        }
        if (isFilter) {//是否开启过滤，默认关闭
            cacheFlag = MessageFilter.repeatFilter(params, request);
            if (!cacheFlag) {//重复推送直接返回
                writeResponse(ctx.channel(), request, ResponseStatus.REPETITION);
                return;
            }
        }
        httpAsyncClient.execute(httpPost, new FutureCallback<HttpResponse>() {
            public void completed(final HttpResponse response) {
                BufferedReader in = null;
                String resultString = "";
                try {
                    InputStream is = response.getEntity().getContent();
                    in = new BufferedReader(new InputStreamReader(is, "UTF-8"));
                    StringBuffer buffer = new StringBuffer();
                    String line = "";
                    while ((line = in.readLine()) != null) {
                        buffer.append(line);
                    }
                    resultString = buffer.toString();
                    logger.info("服务返回信息:{}", resultString);
                    if (JsonUtil.getResponseVo(resultString).getCode() == 0) {
                        writeResponse(ctx.channel(), request, success);
                    } else {

                        writeResponse(ctx.channel(), request, error);
                    }
                } catch (IOException ioe) {
                    logger.error("异步分发异常:{}", ioe.getMessage());
                    throw new RuntimeException(ioe);
                }
                logger.debug(httpPost.getRequestLine() + "->" + response.getStatusLine());
            }

            public void failed(final Exception ex) {
                //TODO 服务报错可以加入通知
//                writeResponse(ctx.channel(), request, error);
                logger.debug(url + "服务报错,进行错误重试");
                //TODO 加入失败集合
                doExcutes(new HttpRequestWrapper(ctx, request, msg, url, success, error));
                logger.debug(httpPost.getRequestLine() + "->错误:" + ex);
            }

            public void cancelled() {
                writeResponse(ctx.channel(), request, error);
                logger.debug(httpPost.getRequestLine() + " cancelled");
            }
        });

    }

    /**
     * 从url中获得信息
     * 此方法未使用,目前使用RequestParse 解析参数
     * getParamByUrl
     *
     * @param
     */
    private static Map<String, String> getParamByUrl(HttpRequest request) {
        Map<String, String> params = new HashMap<>();
        QueryStringDecoder decoderQuery = new QueryStringDecoder(request.getUri());
        Map<String, List<String>> uriAttributes = decoderQuery.parameters();
        for (Map.Entry<String, List<String>> attr : uriAttributes.entrySet()) {
            for (String attrVal : attr.getValue()) {
                logger.debug("URI: " + attr.getKey() + '=' + attrVal + "\r\n");
                params.put(attr.getKey(), attrVal);
            }
        }
        return params;
    }


    /**
     * 处理异步请求
     *
     * @param ctx
     * @param url
     */
    public static void doGetExecute(ChannelHandlerContext ctx, HttpRequest request, HttpObject msg, String url, String success, String error) {
        boolean cacheFlag = true;
        StringBuilder sb = new StringBuilder(url + "?");
        Map<String, String> params = RequestParser.parse(request, msg);
        Set<String> keys = params.keySet();
        for (String key : keys) {
            String value = params.get(key);
            sb.append(key).append("=").append(value).append("&");
            logger.debug("key:{},value:{}", key, value);
        }
        String urls = sb.substring(0, sb.length() - 1);
        if (isFilter) {//是否开启过滤，默认关闭
            cacheFlag = MessageFilter.repeatFilter(urls, request);
            if (!cacheFlag) {//重复推送直接返回
                writeResponse(ctx.channel(), request, ResponseStatus.REPETITION);
                return;
            }
        }
        HttpGet httpget = new HttpGet(url);
        pretreatentHead(request, httpget);
        httpAsyncClient.execute(httpget, new FutureCallback<HttpResponse>() {
            public void completed(final HttpResponse response) {
                BufferedReader in = null;
                String resultString = "";
                System.out.println(response.getStatusLine());
                logger.debug("品类响应长度:{}", response.getHeaders(CONTENT_LENGTH));
                try {
                    InputStream is = response.getEntity().getContent();
                    in = new BufferedReader(new InputStreamReader(is, "UTF-8"));
                    StringBuffer buffer = new StringBuffer();
                    String line = "";
                    while ((line = in.readLine()) != null) {
                        buffer.append(line);
                    }
                    resultString = buffer.toString();
                    logger.info("服务返回信息:{}", resultString);
                    if (JsonUtil.getResponseVo(resultString).getCode() == 0) {
                        writeResponse(ctx.channel(), request, success);
                    } else {
                        writeResponse(ctx.channel(), request, error);
                    }
                } catch (IOException ioe) {
                    logger.error("异步分发异常:{}", ioe.getMessage());
                    throw new RuntimeException(ioe);
                }
                System.out.println(httpget.getRequestLine() + "->" + response.getStatusLine());
            }

            public void failed(final Exception ex) {
                //TODO 服务报错可以加入通知
//                writeResponse(ctx.channel(), request, error);
                System.out.println(urls + "服务报错,进行错误重试");
                //TODO 加入失败集合
                doExcutes(new HttpRequestWrapper(ctx, request, msg, url, success, error));
                System.out.println(httpget.getRequestLine() + "->" + ex);
            }

            public void cancelled() {
                writeResponse(ctx.channel(), request, error);
                System.out.println(httpget.getRequestLine() + " cancelled");
            }
        });
    }

    /**
     * 返回制定的友好提示
     *
     * @param channel
     * @param request
     * @param responseStr
     */
    private static void writeResponse(Channel channel, HttpRequest request, String responseStr) {
        String strVar = responseStr;
        ByteBuf buf = copiedBuffer(strVar.toString(), CharsetUtil.UTF_8);

        boolean close = request.headers().contains(CONNECTION, HttpHeaders.Values.CLOSE, true)
                || request.getProtocolVersion().equals(HttpVersion.HTTP_1_0)
                && !request.headers().contains(CONNECTION, HttpHeaders.Values.KEEP_ALIVE, true);

        //构建响应对象
        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, buf);
        response.headers().set(CONTENT_TYPE, "text/plain; charset=UTF-8");

        if (!close) {
            //构建内容长度
            response.headers().set(CONTENT_LENGTH, buf.readableBytes());
        }
        Set<Cookie> cookies;
        String value = request.headers().get(COOKIE);
        if (value == null) {
            cookies = Collections.emptySet();
        } else {
            cookies = CookieDecoder.decode(value);
        }
        if (!cookies.isEmpty()) {
            // Reset the cookies if necessary.
            for (Cookie cookie : cookies) {
                response.headers().add(SET_COOKIE, ServerCookieEncoder.encode(cookie));
            }
        }
        // 执行完毕回调
        ChannelFuture future = channel.writeAndFlush(response);
        // 写操作完成后关闭连接
        if (close) {
            future.addListener(ChannelFutureListener.CLOSE);
        }
    }

    /**
     * 没有找到默认返回404
     *
     * @param channel
     * @param request
     * @param responseStr ResonseStatus中对应的提示
     */
    public static void defaultResponse(Channel channel, HttpRequest request, String responseStr) {
        String strVar = responseStr;
        ByteBuf buf = copiedBuffer(strVar.toString(), CharsetUtil.UTF_8);
        //连接结束的时候返回true响应信息
        boolean close = request.headers().contains(CONNECTION, HttpHeaders.Values.CLOSE, true)
                || request.getProtocolVersion().equals(HttpVersion.HTTP_1_0)
                && !request.headers().contains(CONNECTION, HttpHeaders.Values.KEEP_ALIVE, true);

        //构建响应对象
        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, buf);
        response.headers().set(CONTENT_TYPE, "text/plain; charset=UTF-8");
        response.setStatus(HttpResponseStatus.NOT_FOUND);
        if (!close) {
            //构建内容长度
            response.headers().set(CONTENT_LENGTH, buf.readableBytes());
        }
        Set<Cookie> cookies;
        String value = request.headers().get(COOKIE);
        if (value == null) {
            cookies = Collections.emptySet();
        } else {
            cookies = CookieDecoder.decode(value);
        }
        if (!cookies.isEmpty()) {
            // Reset the cookies if necessary.
            for (Cookie cookie : cookies) {
                response.headers().add(SET_COOKIE, ServerCookieEncoder.encode(cookie));
            }
        }
        // 执行完毕回调
        ChannelFuture future = channel.writeAndFlush(response);
        // 写操作完成后关闭连接
        if (close) {
            future.addListener(ChannelFutureListener.CLOSE);
        }
    }

    public static void doGetExecutes(HttpRequestWrapper httpRequestWrapper) {
        ChannelHandlerContext ctx=httpRequestWrapper.getCtx();
        HttpRequest request=httpRequestWrapper.getRequest();
        HttpObject msg=httpRequestWrapper.getMsg();
        String url=httpRequestWrapper.getUrl();
        String success=httpRequestWrapper.getSuccess();
        String error=httpRequestWrapper.getError();
        boolean cacheFlag = true;
        StringBuilder sb = new StringBuilder(url + "?");
        Map<String, String> params = RequestParser.parse(request, msg);
        Set<String> keys = params.keySet();
        for (String key : keys) {
            String value = params.get(key);
            sb.append(key).append("=").append(value).append("&");
            logger.debug("key:{},value:{}", key, value);
        }
        String urls = sb.substring(0, sb.length() - 1);
        if (isFilter) {//是否开启过滤，默认关闭
            cacheFlag = MessageFilter.repeatFilter(urls, request);
            if (!cacheFlag) {//重复推送直接返回
                writeResponse(ctx.channel(), request, ResponseStatus.REPETITION);
                return;
            }
        }
        HttpGet httpget = new HttpGet(url);
        pretreatentHead(request, httpget);
        httpAsyncClient.execute(httpget, new FutureCallback<HttpResponse>() {
            public void completed(final HttpResponse response) {
                BufferedReader in = null;
                String resultString = "";
                System.out.println(response.getStatusLine());
                logger.debug("品类响应长度:{}", response.getHeaders(CONTENT_LENGTH));
                try {
                    InputStream is = response.getEntity().getContent();
                    in = new BufferedReader(new InputStreamReader(is, "UTF-8"));
                    StringBuffer buffer = new StringBuffer();
                    String line = "";
                    while ((line = in.readLine()) != null) {
                        buffer.append(line);
                    }
                    resultString = buffer.toString();
                    logger.info("服务返回信息:{}", resultString);
                    if (JsonUtil.getResponseVo(resultString).getCode() == 0) {
                        writeResponse(ctx.channel(), request, success);
                    } else {
                        writeResponse(ctx.channel(), request, error);
                    }
                } catch (IOException ioe) {
                    logger.error("异步分发异常:{}", ioe.getMessage());
                    throw new RuntimeException(ioe);
                }
                System.out.println(httpget.getRequestLine() + "->" + response.getStatusLine());
            }

            public void failed(final Exception ex) {
                //TODO 服务报错可以加入通知
                    writeResponse(ctx.channel(), request, error);
                System.out.println(urls + "服务报错,进行错误重试:"+httpRequestWrapper.getErrorNum());
                System.out.println(httpget.getRequestLine() + "->" + ex);
            }

            public void cancelled() {
                writeResponse(ctx.channel(), request, error);
                System.out.println(httpget.getRequestLine() + " cancelled");
            }
        });
    }

    public static void doPostExecutes(HttpRequestWrapper httpRequestWrapper) {
        logger.debug("===THIS IS HTTP POST===");
        ChannelHandlerContext ctx=httpRequestWrapper.getCtx();
        HttpRequest request=httpRequestWrapper.getRequest();
        HttpObject msg=httpRequestWrapper.getMsg();
        String url=httpRequestWrapper.getUrl();
        String success=httpRequestWrapper.getSuccess();
        String error=httpRequestWrapper.getError();
        boolean cacheFlag = true;
        HttpPost httpPost = new HttpPost(url);
        pretreatentHead(request, httpPost);
        Map<String, String> params = RequestParser.parse(request, msg);
        List<NameValuePair> nvps = new ArrayList<NameValuePair>();
        Set<String> keys = params.keySet();
        for (String key : keys) {
            String value = params.get(key);
            logger.debug("key:{},value:{}", key, value);
            nvps.add(new BasicNameValuePair(key, value));
            try {
                httpPost.setEntity(new UrlEncodedFormEntity(nvps));
            } catch (UnsupportedEncodingException ue) {

            }
        }
        if (isFilter) {//是否开启过滤，默认关闭
            cacheFlag = MessageFilter.repeatFilter(params, request);
            if (!cacheFlag) {//重复推送直接返回
                writeResponse(ctx.channel(), request, ResponseStatus.REPETITION);
                return;
            }
        }
        httpAsyncClient.execute(httpPost, new FutureCallback<HttpResponse>() {
            public void completed(final HttpResponse response) {
                BufferedReader in = null;
                String resultString = "";
                try {
                    InputStream is = response.getEntity().getContent();
                    in = new BufferedReader(new InputStreamReader(is, "UTF-8"));
                    StringBuffer buffer = new StringBuffer();
                    String line = "";
                    while ((line = in.readLine()) != null) {
                        buffer.append(line);
                    }
                    resultString = buffer.toString();
                    logger.info("服务返回信息:{}", resultString);
                    if (JsonUtil.getResponseVo(resultString).getCode() == 0) {
                        writeResponse(ctx.channel(), request, success);
                    } else {
                        writeResponse(ctx.channel(), request, error);
                    }
                } catch (IOException ioe) {
                    logger.error("异步分发异常:{}", ioe.getMessage());
                    throw new RuntimeException(ioe);
                }
                logger.debug(httpPost.getRequestLine() + "->" + response.getStatusLine());
            }

            public void failed(final Exception ex) {
                //TODO 服务报错可以加入通知
                writeResponse(ctx.channel(), request, error);
                logger.debug(url + "服务报错,进行错误重试:"+httpRequestWrapper.getErrorNum());
                //TODO 加入失败集合
                logger.debug(httpPost.getRequestLine() + "->错误:" + ex);
            }

            public void cancelled() {
                writeResponse(ctx.channel(), request, error);
                logger.debug(httpPost.getRequestLine() + " cancelled");
            }
        });

    }
    /**
     *
     * @param httpRequestWrapper
     */
    public static void doExcutes(HttpRequestWrapper httpRequestWrapper) {
       HttpRequest request=httpRequestWrapper.getRequest();
        if (request.getMethod().equals(HttpMethod.POST)) {
            doPostExecutes(httpRequestWrapper);
        } else if (request.getMethod().equals(HttpMethod.GET)) {
            doGetExecutes(httpRequestWrapper);
        }

    }
}
